clients/src/main/java/org/apache/kafka/common/config/provider/ConfigProvider.java (1)

32-35: Clarify when metrics and tags are attached to providers

Tags are added only when the provider is wrapped via Plugins.newConfigProvider and a non-null Metrics is supplied; the class tag is the simple class name. Consider tightening the Javadoc to avoid implying unconditional tagging.

- * <p>Implement {@link org.apache.kafka.common.metrics.Monitorable} to enable the config provider to register metrics.
- * The following tags are automatically added to all metrics registered: <code>config</code> set to
- * <code>config.providers</code>, <code>class</code> set to the ConfigProvider class name,
- * and <code>provider</code> set to the provider name.
+ * <p>Implement {@link org.apache.kafka.common.metrics.Monitorable} to enable the config provider to register metrics
+ * when created via {@code Plugins.newConfigProvider(...)} with a non-null {@code Metrics}.
+ * The following tags are added to all metrics registered: <code>config</code> = <code>config.providers</code>,
+ * <code>class</code> = the provider's simple class name, and <code>provider</code> = the configured provider name.

connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMakerConfig.java (1)

270-287: Avoid null entries in providerPlugins; filter before put and close

newConfigProvider may return null; skip adding those and you won’t rely on Utils.closeQuietly handling nulls. Keeps the map clean and intent clear.

-        Map<String, Plugin<ConfigProvider>> providerPlugins = new HashMap<>();
+        Map<String, Plugin<ConfigProvider>> providerPlugins = new HashMap<>();
         for (String name : providerNames) {
             Plugin<ConfigProvider> configProviderPlugin = plugins.newConfigProvider(
                     this,
                     name,
                     Plugins.ClassLoaderUsage.PLUGINS,
                     null
             );
-            providerPlugins.put(name, configProviderPlugin);
+            if (configProviderPlugin != null) {
+                providerPlugins.put(name, configProviderPlugin);
+            }
         }
         ConfigTransformer transformer = new ConfigTransformer(providerPlugins);
         Map<String, String> transformed = transformer.transform(props).data();
-        providerPlugins.values().forEach(x -> Utils.closeQuietly(x, "config provider plugin"));
+        providerPlugins.values().forEach(p -> Utils.closeQuietly(p, "config provider plugin"));
         return transformed;

clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java (1)

624-633: Add provider name as a Plugin metrics tag for better observability

Include the provider’s logical name in Plugin tags to aid metrics/log correlation.

-                Plugin<ConfigProvider> providerPlugin = Plugin.wrapInstance(provider, null, CONFIG_PROVIDERS_CONFIG);
+                Plugin<ConfigProvider> providerPlugin = Plugin.wrapInstance(
+                        provider,
+                        null,
+                        CONFIG_PROVIDERS_CONFIG,
+                        Collections.singletonMap("name", entry.getKey())
+                );

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java (1)

210-223: Clean up partially created provider plugins if initialization fails

If creating a later provider or constructing WorkerConfigTransformer throws, earlier plugins remain unclosed. Add failure-path cleanup.

     private WorkerConfigTransformer initConfigTransformer() {
         final List<String> providerNames = config.getList(WorkerConfig.CONFIG_PROVIDERS_CONFIG);
-        Map<String, Plugin<ConfigProvider>> providerPluginMap = new HashMap<>();
-        for (String providerName : providerNames) {
-            Plugin<ConfigProvider> configProviderPlugin = plugins.newConfigProvider(
-                config,
-                providerName,
-                ClassLoaderUsage.PLUGINS,
-                metrics.metrics()
-            );
-            providerPluginMap.put(providerName, configProviderPlugin);
-        }
-        return new WorkerConfigTransformer(this, providerPluginMap);
+        Map<String, Plugin<ConfigProvider>> providerPluginMap = new HashMap<>();
+        try {
+            for (String providerName : providerNames) {
+                Plugin<ConfigProvider> configProviderPlugin = plugins.newConfigProvider(
+                        config,
+                        providerName,
+                        ClassLoaderUsage.PLUGINS,
+                        metrics.metrics()
+                );
+                providerPluginMap.put(providerName, configProviderPlugin);
+            }
+            return new WorkerConfigTransformer(this, providerPluginMap);
+        } catch (Throwable t) {
+            providerPluginMap.values().forEach(p -> Utils.closeQuietly(p, "config provider plugin"));
+            throw t;
+        }
     }

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerConfigTransformerTest.java (1)

23-30: Close the transformer in tests to exercise/verify plugin lifecycle

Add @AfterEach to close configTransformer; future-proofs tests if providers acquire resources.

@@
-import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.AfterEach;
@@
     public void setup() {
         configTransformer = new WorkerConfigTransformer(worker, Map.of("test", Plugin.wrapInstance(new TestConfigProvider(), null, "config.providers")));
     }
+
+    @AfterEach
+    public void tearDown() {
+        if (configTransformer != null) {
+            configTransformer.close();
+        }
+    }

Also applies to: 67-71

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfigTransformer.java (1)

46-52: Plugin-wrapped providers: wiring looks correct

Constructor and field migration to Map<String, Plugin> aligns with the Plugin API; passing the plugin map to ConfigTransformer is consistent with the PR direction. Consider guarding against external mutation by copying/wrapping the map.

Apply within this hunk:

-        this.configProviderPlugins = configProviderPlugins;
-        this.configTransformer = new ConfigTransformer(configProviderPlugins);
+        this.configProviderPlugins = java.util.Collections.unmodifiableMap(new java.util.HashMap<>(configProviderPlugins));
+        this.configTransformer = new ConfigTransformer(this.configProviderPlugins);

Confirm ConfigTransformer’s constructor now accepts Map<String, Plugin> to avoid a cross-module signature mismatch.

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java (1)

2904-2927: New test asserts provider metrics exposure — ensure init timing

The assertions read metrics immediately after Worker construction (without worker.start()). If Worker’s provider initialization ever moves to start(), this test could become flaky.

Options:

• Call worker.start() and worker.stop() in this test, or
• Add a small accessor/trigger that initializes provider plugins deterministically for tests.

Verify this test passes consistently across configurations (no reliance on side effects in the constructor).

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/PluginsTest.java (1)

404-412: withPluginMetrics-after-configure ordering is validated

Creating a monitorable provider with Metrics and asserting instance type relies on overridden assertion for order — good. Consider also asserting at least one “plugins” metric exists to strengthen the test.

Example follow-up within this test:

• Inspect new Metrics instance for a MetricName with group "plugins" and class tag CustomMonitorableConfigProvider.

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro